package com.eureka.provider.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.eureka.provider.lock.ZookeeperLock;
import com.eureka.provider.mapper.SeckillGoodsMapper;
import com.eureka.provider.pojo.Order;
import com.eureka.provider.pojo.SeckillGoods;
import com.eureka.provider.pojo.User;
import com.eureka.provider.service.ISeckillGoodsService;
import com.eureka.provider.vo.RespBean;
import com.eureka.provider.vo.RespBeanEnum;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author sam
 * @since 2021-04-15
 */
@Service
@Slf4j
public class SeckillGoodsServiceImpl extends ServiceImpl<SeckillGoodsMapper, SeckillGoods> implements ISeckillGoodsService {


    @Autowired
    Redisson redisson;
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    KafkaTemplate kafkaTemplate;

    /**
     * redisson锁
     * @param user
     * @param goodsId
     * @return
     */
    @Override
    public RespBean doSecKill1(User user, Long goodsId) {
        /**
         * redis主从架构锁失效问题：
         * CAP（一致性、可用性、分区容错性）理论：redis集群满足AP 单机CP zk集群满足CP
         * 原因：redis集群加锁后马上返回客户端结果，然后同步（牺牲一致性）
         *     zk集群先保证半数节点同步成功，再返回加锁结果（牺牲可用性）
         * 选择策略：redis并发高 zk并发较低
         */
        RLock lock = redisson.getLock("goodsLock"+goodsId);
        try {
            //如果加锁失败，while循环自旋加锁
            lock.lock();
            //尝试获取锁，最多等待3秒
            //lock.tryLock(3, TimeUnit.SECONDS);
            int stock=(int)redisTemplate.opsForValue().get("stock:"+goodsId);
            if(stock>0){
                redisTemplate.opsForValue().set("stock:"+goodsId,stock-1);
                log.info("减库存成功");
                Order order=new Order();
                order.setGoodsId(goodsId);
                order.setUserId(user.getId());
                kafkaTemplate.send("order", JSON.toJSONString(order));
                return RespBean.Success();
            }else{
                //log.info("失败");
                return RespBean.Error();
            }
        }catch (Exception e){
            return RespBean.Error();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 秒杀2：redis原子递减
     * @param user
     * @param goodsId
     * @return
     */
    @Override
    public RespBean doSecKill2(User user, Long goodsId) {
        int stock=(int)redisTemplate.opsForValue().get("stock:"+goodsId);
        if(stock<1){
            return RespBean.Error(RespBeanEnum.EMPTY_STOCK);
        }
        Long decrement = redisTemplate.opsForValue().decrement("stock:" + goodsId, 1);
        if(decrement>=0){
            log.info("减库存成功");
            Order order=new Order();
            order.setGoodsId(goodsId);
            order.setUserId(user.getId());
            kafkaTemplate.send("order", JSON.toJSONString(order));
            return RespBean.Success();
        }else{
            return RespBean.Error();
        }
    }

    @Autowired
    SeckillGoodsMapper seckillGoodsMapper;

    /**
     * 秒杀3：zookeeper锁
     * @param user
     * @param goodsId
     * @return
     */
    @Override
    public RespBean doSecKill3(User user, Long goodsId) {
        SeckillGoods goods = seckillGoodsMapper.selectById(goodsId);
        if(goods.getSeckillStock()<1){
            return RespBean.Error(RespBeanEnum.EMPTY_STOCK);
        }
        boolean acquire = ZookeeperLock.acquire(3, TimeUnit.SECONDS);
        if(acquire){
            goods = seckillGoodsMapper.selectById(goodsId);
            goods.setSeckillStock(goods.getSeckillStock()-1);
            seckillGoodsMapper.updateById(goods);
            log.info(RespBeanEnum.SUCCESS.getMsg());
        }else{
            return RespBean.Error();
        }
        return null;
    }


}
